package com.doit.day03;

import com.alibaba.fastjson.JSON;
import com.google.common.hash.BloomFilter;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MyConsumerInterceptor implements ConsumerInterceptor<String,String> {
    private BloomFilter<Long> longBloomFilter;
    public MyConsumerInterceptor(BloomFilter<Long> longBloomFilter){
        this.longBloomFilter  = longBloomFilter;
    }
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        //Map<TopicPartition, List<ConsumerRecord<K, V>>> records
        List<ConsumerRecord<String, String>> consumerRecords = new ArrayList<>();
        Map<TopicPartition, List<ConsumerRecord<String, String>>> topicPartitionListHashMap = new HashMap<>();
        ConsumerRecords<String, String> records1 = new ConsumerRecords<String, String>(topicPartitionListHashMap);
        for (ConsumerRecord<String, String> record : records) {
            String value = record.value();
            EventLog eventLog = JSON.parseObject(value, EventLog.class);
            boolean flag1 = longBloomFilter.mightContain(eventLog.getGuid());
            //代表是true
            if (flag1){
                //把字段设置成0
                eventLog.setIsNew(0);
            }else {
                //不存在，代表是新用户
                eventLog.setIsNew(1);
                //把新用户标注到布隆过滤器中
                longBloomFilter.put(eventLog.getGuid());
                //标注到bitmap中，下次去判断的时候就是一个老用户
            }
            String res = JSON.toJSONString(eventLog);
            TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
            ConsumerRecord<String, String> stringStringConsumerRecord = new ConsumerRecord<>(record.topic(), record.partition(), record.offset(), record.key(), res);
            consumerRecords.add(stringStringConsumerRecord);
            topicPartitionListHashMap.put(topicPartition,consumerRecords);
        }
        return records1;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
